package io.summer.gadget.channel.core.impl;

import io.summer.basis.excepts.AppException;
import io.summer.basis.tool.HttpTool;
import io.summer.basis.tool.LambdaTool;
import io.summer.basis.tool.MyReflectionTool;
import io.summer.gadget.channel.SummerChannel;
import io.summer.gadget.channel.core.ChannelConst;
import io.summer.gadget.channel.core.ChannelRequest;
import io.summer.gadget.channel.core.Protocol;
import io.summer.gadget.channel.core.ServerChooser;
import io.summer.gadget.channel.core.ServiceRegistry;
import io.summer.gadget.channel.core.TopicListener;
import io.summer.gadget.channel.core.TopicUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.retry.support.RetryTemplate;

import java.lang.reflect.Method;
import java.lang.reflect.Type;
import java.util.Collections;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

/**
 * @author ml
 */
@Slf4j
public class SummerChannelImpl implements SummerChannel {
  @Autowired(required = false)
  protected ServiceRegistry serviceRegistry;

  @Autowired
  protected TopicUtil topicUtil;

  @Autowired
  private RetryTemplate retryTemplate;

  @Autowired
  protected ServerChooser serverChooser;

  @Autowired
  protected Protocol protocol;

  private static final String ERROR_MSG = "publish error: ";

  protected final Map<String, List<ChannelConsumer>> topicFunctionMap = new HashMap<>();

  private Type[] getGenericTypes(Class<?> clazz) {
    Type[] pTypes = MyReflectionTool.getGenericTypesOfClass(clazz, 0);
    if (pTypes.length == 0) {
      Method[] methods = clazz.getDeclaredMethods();
      Method m = methods[0];
      pTypes = new Type[ChannelConsumer.HANDLER_SIG_TYPES];
      pTypes[ChannelConsumer.RETURN_TYPE] = m.getGenericReturnType();
      pTypes[ChannelConsumer.INPUT_TYPE] = m.getGenericParameterTypes()[0];
    }

    if (pTypes.length != ChannelConsumer.HANDLER_SIG_TYPES) {
      log.error("wrongs typed arguments count found: {}|{}", pTypes.length, clazz.getName());
      return null;
    }
    return pTypes;
  }

  private void checkTopicConsumer(Object currentConsumer, List<ChannelConsumer> registered) {
    if (registered.stream().anyMatch(p -> p.consumer.equals(currentConsumer))) {
      throw new AppException("can't subscribe topic repeatedly");
    }
  }

  @Override
  public <T, R> void subscribe(String topic, TopicListener<T, R> consumer) {
    if (serviceRegistry == null) {
      log.warn("[subscribe] no registryService, add service.self.domain and enable discover to register self.");
      return;
    }
    List<ChannelConsumer> lists = topicFunctionMap.getOrDefault(topic, new LinkedList<>());
    checkTopicConsumer(consumer, lists);
    Type[] type = getGenericTypes(consumer.getClass());
    ChannelConsumer po = new ChannelConsumer(type, consumer);
    lists.add(po);
    topicFunctionMap.putIfAbsent(topic, lists);
    serviceRegistry.subscribe(topic);
    log.info("subscribe topic: {}, handler: {}", topic, po);
  }

  private byte[] doSendToServer(String serverName, String receiver, String topic, byte[] data) {
    List<String> servers = serviceRegistry.getServers(serverName);
    if (servers.isEmpty()) {
      log.error("no server available: {}", serverName);
      return new byte[0];
    }

    String uri = serverChooser.choose(serverName, servers);
    log.info("sending to: {}", uri);
    ChannelRequest request = new ChannelRequest();
    request.setData(data);
    request.setFrom(serviceRegistry.getAppName());
    request.setTopic(topic);

    byte[] response = HttpTool.post(uri + receiver, request);
    if (log.isDebugEnabled()) {
      log.debug("process topic: {}, server: {}, uri: {}, response: {}", topic, serverName, uri, response);
    }
    return response;
  }

  private byte[] sendMessage(String topic, String receiver, byte[] data) {
    if (serviceRegistry == null) {
      log.warn("[sendMessage] no registryService, add service.self.domain and enable discover to register self.");
      throw new AppException("serviceRegistry is null, not enabled?");
    }

    List<String> servers = topicUtil.getSubscribedAppsForTopic(topic);
    log.info("about send message [{}] to: {}", topic, servers);
    byte[] response = null;
    for (String server : servers) {
      response = retryTemplate.execute(arg -> doSendToServer(server, receiver, topic, data));
    }
    return response;
  }

  @Override
  public <T> void publish(String topic, T data) {
    try {
      sendMessage(topic, ChannelConst.RECEIVER, protocol.serialize(data));
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
    }
  }

  @Override
  public void publish(String topic, byte[] data) {
    try {
      sendMessage(topic, ChannelConst.RECEIVER, data);
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
    }
  }

  @Override
  public <T> void publishAsync(String topic, T data) {
    try {
      sendMessage(topic, ChannelConst.RECEIVER_ASYNC, protocol.serialize(data));
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
    }
  }

  @Override
  public void publishAsync(String topic, byte[] data) {
    try {
      sendMessage(topic, ChannelConst.RECEIVER_ASYNC, data);
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
    }
  }

  @Override
  public List<ChannelConsumer> getChannelConsumers(String topic) {
    return topicFunctionMap.getOrDefault(topic, Collections.emptyList());
  }

  @Override
  public <T, R> R publish(String topic, Class<R> responseType, T data) {
    return publish(topic, (Type) responseType, data);
  }

  @Override
  public <R> R publish(String topic, Class<R> responseType, byte[] data) {
    return publish(topic, (Type) responseType, data);
  }

  @Override
  public <T, R> R publish(String topic, Type responseType, T data) {
    try {
      return publish(topic, responseType, protocol.serialize(data));
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
      return null;
    }
  }

  @Override
  @SuppressWarnings("unchecked")
  public <R> R publish(String topic, Type responseType, byte[] data) {
    try {
      byte[] response = sendMessage(topic, ChannelConst.RECEIVER, data);
      if (response != null && response.length > 0) {
//        if (responseType.equals(String.class)) {
//          return (R) new String(response);
//        }
        return protocol.deserialize(response, responseType);
      }
    } catch (Exception e) {
      log.error(ERROR_MSG, e);
    }
    return null;
  }
}
